package com.chb.storm.kafkatopology;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.common.utils.AdDateFormatUtils;
import com.common.utils.MobileChbLogConstant;
import com.common.utils.TableUtils;
import com.mobilechb.bean.DriveAllBean;
import com.mobilechb.bean.GpsInfoBean;
import com.mobilechb.bean.IdentificationBean;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
/**
 * 建表：create 'storm_ns:testKafkaToHbase', { NAME => 'd', COMPRESSION => 'snappy'}
 * @author Administrator
 *
 */
public class KafkaTopologyHbaseLogParseBolt implements IBasicBolt {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(KafkaTopologyHbaseLogParseBolt.class);
	
	private static final byte[] FAMILY_C = Bytes.toBytes("d");
	
	private static final byte[] LNG_Q = Bytes.toBytes("lng");
	
	private static final byte[] LAT_Q = Bytes.toBytes("lat");
	
	private static final byte[] T_Q = Bytes.toBytes("t");
	
	private static final byte[] S_Q = Bytes.toBytes("s");
	
	private static final byte[] TYPE_Q = Bytes.toBytes("type");
	
	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		
	}

	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		String logStr = input.getString(0);
		if (StringUtils.isBlank(logStr))
			return;
		
		String[] logStrArr = StringUtils.split(logStr, "=");
		int len = ArrayUtils.getLength(logStrArr);
		if (len < 2)
		{
			return;
		}
		
		String logPrefix = logStrArr[0];
		String logContent = logStrArr[1];
		//截取log类型
		String logType = StringUtils.substring(logPrefix, 24);
		if (!StringUtils.equalsIgnoreCase(MobileChbLogConstant.DRIVE_ALL, logType))
		{
			return;
		}
		
		DriveAllBean driveAllBean = null;
		try {
			driveAllBean = JSON.parseObject(logContent, DriveAllBean.class);
		} catch (Exception e) {
			return;
		}
		
		IdentificationBean iden = driveAllBean.getIdentification();
		String userId = iden.getUserId();
		if (StringUtils.isBlank(userId))
		{
			return;
		}
		String fixedLenUserId = insureUserId(userId);
		
		String habitId = iden.getHabitId();
		if (StringUtils.isBlank(habitId)
				|| habitId.length() > 11)
		{
			return;
		}
		
		long currentTs = System.currentTimeMillis()/1000;
		String gpsType = driveAllBean.getGpsType();
		if (StringUtils.isBlank(gpsType))
		{
			return;
		}
		
		ArrayList<Put> gpsList = new ArrayList<Put>();
		GpsInfoBean[] gpsInfos = driveAllBean.getGpsInfos();
		for (GpsInfoBean gpsInfo : gpsInfos)
		{
			String lng = gpsInfo.getLng();
			String lat = gpsInfo.getLat();
			String ts = gpsInfo.getT();
			String speed = gpsInfo.getS();
			
			if (StringUtils.isBlank(ts))
			{
				continue;
			}
			String rowKeyTs = AdDateFormatUtils.formatServerTimeStr(ts);
			String rowkey = fixedLenUserId + "_" + rowKeyTs;
			
			Put gpsPut = new Put(Bytes.toBytes(rowkey), currentTs);
			gpsPut.setDurability(Durability.SYNC_WAL);
			gpsPut.addColumn(FAMILY_C, LNG_Q, Bytes.toBytes(lng));
			gpsPut.addColumn(FAMILY_C, LAT_Q, Bytes.toBytes(lat));
			gpsPut.addColumn(FAMILY_C, T_Q, Bytes.toBytes(ts));
			gpsPut.addColumn(FAMILY_C, S_Q, Bytes.toBytes(speed));
			gpsPut.addColumn(FAMILY_C, TYPE_Q, Bytes.toBytes(gpsType));
			gpsList.add(gpsPut);
		}
		
		try(
			Table table = TableUtils.getTable("storm_ns:testKafkaToHbase");
		)
		{
			Object[] results = new Object[gpsList.size()];
			table.batch(gpsList, results);
		} catch (IOException | InterruptedException e) {
			LOGGER.error("", e);
		}
	}
	
	@Override
	public void cleanup() {
		
	}
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

	/**
	 * 如果userID没有11位，则补齐到11位，并返回
	 * @param userId
	 * @return
	 */
	private static String insureUserId(String userId)
	{
		while (StringUtils.length(userId) < 11)
		{
			userId = "0" + userId;
		}
		return userId;
	}
	
}
